package stormdemo;

import java.util.Map;
import java.util.Random;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

public class SentenceSpout extends BaseRichSpout {
	private static final long serialVersionUID = -7807389313960632180L;
	private SpoutOutputCollector collector;
	private String[] sentences = {"my dog has fleas","i like cold beverages"};
	
	@Override
	public void nextTuple() {
		this.collector.emit(new Values(sentences[new Random().nextInt(sentences.length)]));
	}

	@Override
	public void open(@SuppressWarnings("rawtypes") Map config, TopologyContext context, SpoutOutputCollector collector) {
		this.collector = collector;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("sentence"));
	}

}
